package com.surfilter.massdata.spark.task.daystat2;

import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.Map;
import java.util.TreeMap;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.spark.Accumulator;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.api.java.function.VoidFunction;
import org.apache.spark.broadcast.Broadcast;
import org.apache.spark.sql.DataFrame;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SQLContext;

import com.act.sparkanalyz.service.impl.SparkService.OutQueueEntity;
import com.act.sparkanalyz.task.ISparkTask;
import com.surfilter.massdata.spark.bean.SAN011;
import com.surfilter.massdata.spark.util.CommonUtils;
import com.surfilter.massdata.spark.util.DateUtil;

/**
 * 访问全球网站情况
 * 本省网民访问全球的网站情况（本省网民是指CIP在BR2002中存在）
 */
public class WebSiteVisitsGlobalTask_New implements ISparkTask{
	
	private static final Log log = LogFactory.getLog(WebSiteVisitsGlobalTask_New.class);
	private static final long serialVersionUID = 1L;
	private String outName;
	
	@Override
	public List<OutQueueEntity> execute(Map<String, DataFrame> dataFrames,Map<String, String> commandMap) {
		long start=System.currentTimeMillis();
		log.info("==============WebSiteVisitsGlobalTask begin====================");
		List<OutQueueEntity> outList = new ArrayList<OutQueueEntity>();
		try{
			DataFrame cipFrame = dataFrames.get("dws_hour_san011");
			JavaRDD<Row> cipRDD = cipFrame.toJavaRDD();
			SQLContext sqlContext = cipFrame.sqlContext();
			JavaSparkContext sc = new JavaSparkContext(sqlContext.sparkContext());
			
			
			
	        String dayStr = commandMap.get("-d");
	        Date date = DateUtil.getExecDate(DateUtil.getYesterday(),dayStr);
			
			List<SAN011> javaRDD = createRdd( cipFrame, date);
	    	DataFrame df = sqlContext.createDataFrame(javaRDD, SAN011.class);
	    	
			outList.add(new OutQueueEntity(outName, df));
			
			CommonUtils.deleteTaskTableData("SAN011", date, "WEBSITE_COUNT_SUM", 0, "day");
		}catch(Exception e){
			if(log.isErrorEnabled()){
				log.error(e.getMessage(),e);
			}
		}
		long end=System.currentTimeMillis();
		double min=(end-start)*1.0/(1000*60);
		System.out.println("WebSiteVisitsGlobalTask_New:exectime: "+min+" min............");
		return outList;
	}

	private List<SAN011> createRdd( DataFrame cipFrame,
			Date date) {
		String sql=" select sum(count_value)as count,BUSS_VALUE  from dws_hour_san011_temp where  BUSS_TYPE='WEBSITE_COUNT_SUM' and sta_range=0 GROUP BY BUSS_VALUE";
		DataFrame DF = cipFrame.sqlContext().sql(sql);
		List<Row> list = DF.toJavaRDD().collect();
		List <SAN011> sa_list =new ArrayList<SAN011>();
		for(Row line:list){
			SAN011 san=new SAN011();
			String count=line.getAs("count").toString();
			String dip_flag=line.getAs("BUSS_VALUE").toString();
			san.setYear(DateUtil.getCurrentYear(date));
			san.setHalf_year(DateUtil.getHalfYear(date));
			san.setQuarter(DateUtil.getQuarter(date));
			san.setMonth(DateUtil.getCurrentMonth(date));
			san.setWeek(DateUtil.getCurrentWeek(date));
			san.setDay(DateUtil.getCurrentDay(date));
			san.setBuss_type("WEBSITE_COUNT_SUM");
			san.setBuss_value(dip_flag);
			san.setCount_value(Long.parseLong(count));
			san.setSta_range(0);
			sa_list.add(san);
		}
		return sa_list;
	}
}
